package com.test.service;

import com.test.Utils.StockTradeMapper;
import com.test.entity.SellAndBuy;
import com.test.entity.StockTrade;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.Properties;

@Slf4j
public class StockSellAndBuy {
    public static int number = 0;
    public static FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
            .setHost("192.168.43.150")
            .setPort(6379)
            .build();

    public static void createDataStream(StreamExecutionEnvironment env) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.88.135:19092,192.168.88.135:29092,192.168.88.135:39092");
        properties.setProperty("group.id", "stock-sell-buy");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "stock-trades-",
                new SimpleStringSchema(),
                properties
        );
        consumer.setStartFromEarliest();

        // 创建数据流并添加水印
        DataStream<StockTrade> tradeStream = env
                .addSource(consumer)
                .map(new StockTradeMapper())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<StockTrade>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner((event, timestamp) -> event.getTime())
                );

        // 当天和每一分钟
        tradeStream
                .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(1000)))
                .trigger(EventTimeTrigger.create())
                .aggregate(new AllTradeAggregator(), new AllTradeResultFunction("1s"))
                .addSink(new AccumulatingRedisSink(redisConfig, "1s"));
        //每小时的数据
//        tradeStream
//                .keyBy(StockTrade::getStockCode)
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
//                .trigger(EventTimeTrigger.create())
//                .aggregate(new TradeAggregator(), new TradeResultFunction("1h"))
//                .addSink(new AccumulatingRedisSink(redisConfig, "1h"));

        log.info("Stock sell and buy analysis configured");
    }

    // 自定义Sink函数
    public static class AccumulatingRedisSink extends RichSinkFunction<SellAndBuy> {
        private transient Jedis jedis;
        private final FlinkJedisPoolConfig config;
        private String type;

        public AccumulatingRedisSink(FlinkJedisPoolConfig config, String type) {
            this.config = config;
            this.type = type;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            try {
                super.open(parameters);
                jedis = new Jedis(config.getHost(), config.getPort());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void close() throws Exception {
            if (jedis != null) {
                jedis.close();
            }
            super.close();
        }

        @Override
        public void invoke(SellAndBuy value, Context context) throws Exception {
            //redis中以列表存储，键为当前时间戳，值为下面一样的数据
            double totalBuyAmount;
            double totalSellAmount;
            double totalBuyVolumeCount;
            double totalSellVolumeCount;
            //当天数据（一分钟累计）
            if (type.equals("1s")) {
                String existingValue = jedis.lindex("2:", -1);

                if (existingValue != null) {
                    String[] parts = existingValue.split(":");
                    totalBuyAmount = Double.parseDouble(parts[0]);
                    totalSellAmount = Double.parseDouble(parts[1]);
                    totalBuyVolumeCount = Double.parseDouble(parts[2]);
                    totalSellVolumeCount = Double.parseDouble(parts[3]);
                } else {
                    System.out.println("我是null");
                    totalBuyAmount = 0.0;
                    totalSellAmount = 0.0;
                    totalBuyVolumeCount = 0L;
                    totalSellVolumeCount = 0L;
                }

                totalBuyAmount += Double.parseDouble(value.getBuyAmount());
                totalSellAmount += Double.parseDouble(value.getSellAmount());
                totalBuyVolumeCount += Double.parseDouble(value.getBuyVolumeCount());
                totalSellVolumeCount += Double.parseDouble(value.getSellVolumeCount());

                jedis.rpush("2:", totalBuyAmount + ":" + totalSellAmount + ":" + totalBuyVolumeCount + ":" + totalSellVolumeCount);
            }
        }
    }

    // 定义聚合函数
    public static class AllTradeAggregator implements AggregateFunction<StockTrade, TradeAccumulator, TradeAccumulator> {
        @Override
        public TradeAccumulator createAccumulator() {
            return new TradeAccumulator();
        }

        @Override
        public TradeAccumulator add(StockTrade value, TradeAccumulator accumulator) {
            if ("买入".equals(value.getTradeType())) {
                accumulator.totalBuyAmount += value.getTradeVolume();
                accumulator.totalBuyVolumeCount++;
            } else if ("卖出".equals(value.getTradeType())) {
                accumulator.totalSellAmount += value.getTradeVolume();
                accumulator.totalSellVolumeCount++;
            }

            return accumulator;
        }

        @Override
        public TradeAccumulator getResult(TradeAccumulator accumulator) {
            return accumulator;
        }

        @Override
        public TradeAccumulator merge(TradeAccumulator a, TradeAccumulator b) {
            a.totalBuyAmount += b.totalBuyAmount;
            a.totalSellAmount += b.totalSellAmount;
            a.totalBuyVolumeCount += b.totalBuyVolumeCount;
            a.totalSellVolumeCount += b.totalSellVolumeCount;
            return a;
        }
    }
    public static class TradeAccumulator {
        public static int type = 0;
        public String stockCode;
        public double totalBuyAmount = 0.0;
        public double totalSellAmount = 0.0;
        public long totalBuyVolumeCount = 0L;
        public long totalSellVolumeCount = 0L;

        public TradeAccumulator() {
        }

        public TradeAccumulator(String stockCode) {
            this.stockCode = stockCode;
        }
    }

    // 定义结果函数
    public static class AllTradeResultFunction implements AllWindowFunction<TradeAccumulator, SellAndBuy, TimeWindow> {
        private final String windowType;

        public AllTradeResultFunction(String windowType) {
            this.windowType = windowType;
        }

        @Override
        public void apply(TimeWindow window, Iterable<TradeAccumulator> input, Collector<SellAndBuy> out) throws Exception {
            double totalBuyAmount = 0.0;
            double totalSellAmount = 0.0;
            long totalBuyVolumeCount = 0L;
            long totalSellVolumeCount = 0L;

            for (TradeAccumulator accumulator : input) {
                totalBuyAmount += accumulator.totalBuyAmount;
                totalSellAmount += accumulator.totalSellAmount;
                totalBuyVolumeCount += accumulator.totalBuyVolumeCount;
                totalSellVolumeCount += accumulator.totalSellVolumeCount;
            }

            if (StockSellAndBuy.number < 20) {
                System.out.println("Processing window for window type: " + windowType + ", window: [" + window.getStart() + " - " + window.getEnd() + "], total buy amount: " + totalBuyAmount + ", total sell amount: " + totalSellAmount + ", total buy volume count: " + totalBuyVolumeCount + ", total sell volume count: " + totalSellVolumeCount + "  这是第" + StockSellAndBuy.number + "次");
                StockSellAndBuy.number++;
            }

            out.collect(new SellAndBuy("all", String.valueOf(totalBuyAmount), String.valueOf(totalSellAmount), String.valueOf(totalBuyVolumeCount), String.valueOf(totalSellVolumeCount)));
        }
    }
}
